package com.song.storm_demo.chap_13;

import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.coordination.BatchOutputCollector;
import backtype.storm.drpc.LinearDRPCTopologyBuilder;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseBatchBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class ReachTopology {

	
	public static  HashMap<String, List<String>> TWEETERS = new HashMap<String, List<String>>(){
		{
			put("foo.com/blog/1", Arrays.asList("sally","bob","tim","george","nathan"));
			put("engineering.twitter.com/blog/5", Arrays.asList("adam","david","sally","nathan"));
			put("tech.backtype.com/blog/123", Arrays.asList("tim","mike","john"));
		}
	};
	
	public static  Map<String,List<String>> FOLLOWERS_DB = new HashMap<String, List<String>>(){
		{
			put("sally", Arrays.asList("bob","tim","alice","adam","jim","chris","jai"));
			put("bob", Arrays.asList("sally","nathan","jim","mary","david","vivian"));
			put("tim", Arrays.asList("alex"));
			put("nathan", Arrays.asList("sally","bob","adam","harry","chris","vivian","emily","jordan"));
			put("adam", Arrays.asList("david","carissa"));
			put("mike", Arrays.asList("john","bob"));
			put("john", Arrays.asList("alice","nathan","jim","mike","bob"));
		}
	};
	
	public static class GetTweeters extends BaseBasicBolt
	{

		@Override
		public void execute(Tuple tuple	, BasicOutputCollector collector) {
			// TODO Auto-generated method stub
			Object id = tuple.getValue(0);
			String url = tuple.getString(1);
			List<String> tweeters = TWEETERS.get(url);
			if(tweeters != null)
			{
				for(String tweeter :tweeters)
				{
					collector.emit(new Values(id, tweeter));
				}
			}
		}

		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			// TODO Auto-generated method stub
			declarer.declare(new Fields("id","tweeter"));
		}
		
	}
	
	public static class GetFollowers extends BaseBasicBolt
	{

		@Override
		public void execute(Tuple tuple, BasicOutputCollector collector) {
			Object id = tuple.getValue(0);
			String tweeter = tuple.getString(1);
			List<String> followers = FOLLOWERS_DB.get(tweeter);
			if(followers != null)
			{
				for(String follower :followers)
				{
					collector.emit(new Values(id, follower));
				}
			}
			
		}

		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			declarer.declare(new Fields("id","follower"));
		}
		
	}
	
	
	public static class PartialUniquer extends BaseBatchBolt
	{
		
		BatchOutputCollector _collector;
		Object _id;
		Set<String> _followers = new HashSet<String>();
		@Override
		public void execute(Tuple tuple) {
			// TODO Auto-generated method stub
			_followers.add(tuple.getString(1));
		}
		@Override
		public void finishBatch() {
			_collector.emit(new Values(_id , _followers.size()));
		}
		@Override
		public void prepare(Map map, TopologyContext context,
				BatchOutputCollector collector, Object id) {
			// TODO Auto-generated method stub
			_collector  = collector;
			_id = id;
		}
		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			// TODO Auto-generated method stub
			declarer.declare(new Fields("id","partial-count"));
		}

	}
	
	public static class CountAggregator extends BaseBatchBolt
	{
		
		BatchOutputCollector _collector;
		Object _id;
		int _count = 0;
		
		@Override
		public void execute(Tuple tuple) {
			// TODO Auto-generated method stub
			_count += tuple.getInteger(1);
		}

		@Override
		public void finishBatch() {
			// TODO Auto-generated method stub
			_collector.emit(new Values(_id,_count));
		}

		@Override
		public void prepare(Map map, TopologyContext context,
				BatchOutputCollector collector, Object id) {
			_collector = collector;
			_id = id;
		}

		@Override
		public void declareOutputFields(OutputFieldsDeclarer declarer) {
			declarer.declare(new Fields("id","reach"));
		}
	}
	
	public static LinearDRPCTopologyBuilder init(){
		
		LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach");
		builder.addBolt(new GetTweeters(),4);
		builder.addBolt(new GetFollowers(),12).shuffleGrouping();
		builder.addBolt(new PartialUniquer()).fieldsGrouping(new Fields("id","follower"));
		builder.addBolt(new CountAggregator(),3).fieldsGrouping(new Fields("id"));
		return builder;
	}
	
	public static void main(String[] args)
	{
		LinearDRPCTopologyBuilder builder = init();
		
		Config conf = new Config();
		conf.setMaxTaskParallelism(3);
		LocalDRPC drpc = new LocalDRPC();
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("reach-drpc", conf, builder.createLocalTopology(drpc));
		
		String[] urlsToTry = new String[]{"foo.com/blog/1","engineering.twitter.com/blog/5"};
		
		for(String url :urlsToTry)
		{
			System.out.println("Reach of "+ url+" : "+drpc.execute("reach", url));
		}
		cluster.shutdown();
		drpc.shutdown();
		
	}
	
}
